package com.atguigu.market_analysis

import java.util
import java.util.Map

import org.apache.flink.api.common.state.{MapState, MapStateDescriptor, ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import scala.collection.mutable

/**
 * 市场营销商业指标统计分析
 * 统计App点击量、APP下载量(分渠道)
 *
 * Project: UserBehaviorAnalysis
 * Package: com.atguigu.market_analysis
 * Version: 1.0
 *
 * Created by  WangJX  on 2019/12/12 16:50
 */
object AppMarketingByChannel {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val value: DataStream[String] = env.addSource(new MySource())
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[MarketingUserBehavior](Time.seconds(3)) {
        override def extractTimestamp(element: MarketingUserBehavior): Long = element.timestamp
      })
      .keyBy(_.channel)
      .timeWindow(Time.seconds(20), Time.seconds(5))
      .process(new MyChannelProcess())
      .keyBy(_.endTime)
      .process(new MyResultProcess())

    value.print("output")

    env.execute("AppMarketingByChannel job")
  }
}

case class BehaviorTypesChannelCount(
                                      channel: String,
                                      endTime: Long,
                                      map: mutable.HashMap[String, mutable.HashMap[String, Long]]
                                    )

class MyChannelProcess() extends ProcessWindowFunction[MarketingUserBehavior, BehaviorTypesChannelCount, String, TimeWindow] {

  //  lazy val mapState: MapState[String, Long] = getRuntimeContext.getMapState(new MapStateDescriptor[String, Long]("map_state", classOf[String], classOf[Long]))
  val map: mutable.HashMap[String, mutable.HashMap[String, Long]] = new mutable.HashMap[String, mutable.HashMap[String, Long]]()

  override def process(key: String, context: Context, elements: Iterable[MarketingUserBehavior], out: Collector[BehaviorTypesChannelCount]): Unit = {

    val iterator: Iterator[MarketingUserBehavior] = elements.iterator

    while (iterator.hasNext) {
      val userBehavior: MarketingUserBehavior = iterator.next()

      //循环获得渠道
      val channel: String = userBehavior.channel

      val option = map.get(channel)
      if (option != None) {
        val bMap = option.get
        //区分点击事件
        val count: Option[Long] = bMap.get(userBehavior.behavior)
        if (count != None) {
          val value: Long = count.get
          bMap.put(userBehavior.behavior, value + 1)
        } else {
          bMap.put(userBehavior.behavior, 1)
        }
      } else {
        map.put(channel, new mutable.HashMap[String, Long]())
      }
    }

    out.collect(BehaviorTypesChannelCount(key, context.window.getEnd, map))
    map.clear()
  }
}

class MyResultProcess() extends KeyedProcessFunction[Long, BehaviorTypesChannelCount, String] {

  lazy val mapState: MapState[String, mutable.HashMap[String, Long]] = getRuntimeContext.getMapState(new MapStateDescriptor[String, mutable.HashMap[String, Long]]("map_state", classOf[String], classOf[mutable.HashMap[String, Long]]))

  override def processElement(value: BehaviorTypesChannelCount, ctx: KeyedProcessFunction[Long, BehaviorTypesChannelCount, String]#Context, out: Collector[String]): Unit = {

    val map: mutable.HashMap[String, mutable.HashMap[String, Long]] = value.map

    for (elem <- map) {

      val channel: String = elem._1
      val iterator = elem._2.iterator

      while (iterator.hasNext) {
        //数据中的点击事件和点击量
        val tuple: (String, Long) = iterator.next()


        //获取状态中的值
        val stringToLong: mutable.HashMap[String, Long] = mapState.get(channel)
        if (stringToLong != null) {
          val option: Option[Long] = stringToLong.get(tuple._1)
          if (option != None) {
            val count: Long = option.get
            //更新最终数据
            stringToLong.put(tuple._1, count + tuple._2)
          } else {
            stringToLong.put(tuple._1, tuple._2)
          }
        } else {
          val hashMap = new mutable.HashMap[String, Long]()
          hashMap.put(tuple._1, tuple._2)
          mapState.put(channel, hashMap)
        }
      }
    }

    ctx.timerService().registerEventTimeTimer(value.endTime + 1)
  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, BehaviorTypesChannelCount, String]#OnTimerContext, out: Collector[String]): Unit = {

    val sb: StringBuilder = new StringBuilder()

    sb.append("===============================\n")
      .append("关闭时间：").append(timestamp - 1).append("\n")


    val iterator: util.Iterator[Map.Entry[String, mutable.HashMap[String, Long]]] = mapState.iterator()
    while (iterator.hasNext){
      val value: Map.Entry[String, mutable.HashMap[String, Long]] = iterator.next()
      sb.append("****************** channel:").append(value.getKey).append(" ******************").append("\n")
      val map: mutable.HashMap[String, Long] = value.getValue
      for (elem <- map) {
        sb.append("-------------------\n")
        sb.append("behavior:").append(elem._1)
          .append(" count:").append(elem._2)
          .append("\n")
      }
      sb.append("\n")
    }


    mapState.clear()

    out.collect(sb.toString())


  }
}